package com.translation.tasks;


import lombok.SneakyThrows;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.src.FileSource;
import org.apache.flink.connector.file.src.reader.TextLineInputFormat;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.util.Collector;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.Stream;


public class TranslationTasksTest {


    public static void main(String[] args) throws Exception {

        // 创建一个配置对象
        Configuration config = new Configuration();
//
//        // 设置网络缓冲区的最小数量
//        config.setLong("taskmanager.network.memory.min", 100 * 1024 * 1024); // 10MB
//        // 设置网络缓冲区的最大数量
//        config.setLong("taskmanager.network.memory.max", 200 * 1024 * 1024); // 200MB
//        // 设置网络缓冲区的大小
//        config.setLong("taskmanager.memory.network.buffer.size", 32 * 1024); // 32KB

        // 使用配置创建StreamExecutionEnvironment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(config);

        // 假设你有方法来获取目录下所有txt文件的绝对路径
        List<String> filePaths = getFilePaths("D:\\translation_tasks\\data", "txt");
        final ExecutorService executorService = Executors.newFixedThreadPool(10);
        env.addSource(new RichSourceFunction<Tuple2<String, List<String>>>() {
            @Override
            public void run(SourceContext<Tuple2<String, List<String>>> sourceContext) throws Exception {
                int index = 1;
                for (String filePath : filePaths) {
                    List<String> collect = new ArrayList<>();
                    try(BufferedReader bufferedReader = new BufferedReader(new FileReader(filePath));) {
                        collect = bufferedReader.lines().collect(Collectors.toList());
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    System.out.println("正在读取文件：" + filePath+"  文件大小(行数)： "+collect.size()+"   第"+index+"个文件");
                    sourceContext.collect(new Tuple2<>(new File(filePath).getName(), collect));
                    index ++;
                }
            }

            @Override
            public void cancel() {

            }
        }).setParallelism(1).flatMap(new FlatMapFunction<Tuple2<String, List<String>>, Tuple2<String, List<String>>>() {

            /**
             * 将给定的List随机切分为多个子列表，每个子列表的大小在minSize和maxSize之间。
             *
             * @param originalList 原始列表
             * @param minSize 子列表的最小大小
             * @param maxSize 子列表的最大大小
             * @return 包含随机切分得到的子列表的列表
             */
            public List<List<String>> splitListRandomly(List<String> originalList, int minSize, int maxSize) {
                if (originalList == null || originalList.isEmpty() || minSize <= 0 || maxSize <= 0 || minSize > maxSize) {
                    throw new IllegalArgumentException("Invalid arguments");
                }

                List<List<String>> result = new ArrayList<>();
                Random rand = new Random();
                int remaining = originalList.size();
                while (remaining > 0) {
                    int chunkSize = rand.nextInt((maxSize - minSize) + 1) + minSize;
                    chunkSize = Math.min(chunkSize, remaining); // 确保最后一个子列表不会超出原列表的边界
                    List<String> sublist = new ArrayList<>(originalList.subList(0, chunkSize));
                    originalList.subList(0, chunkSize).clear(); // 移除已处理的部分
                    result.add(sublist);
                    remaining -= chunkSize;
                }
                return result;
            }
            @Override
            public void flatMap(Tuple2<String, List<String>> stringListTuple2, Collector<Tuple2<String, List<String>>> collector) throws Exception {
                List<List<String>> splitLists = splitListRandomly(stringListTuple2.f1, 130, 180);
                int sizzz = 0;
                for (List<String> sublist : splitLists) {
                    sizzz += sublist.size();
                }
                if(sizzz == stringListTuple2.f1.size()){
                    System.out.println("长度校验通过 sizzz = " + sizzz);
                }

                for (List<String> sublist : splitLists) {
                    collector.collect(new Tuple2<String, List<String>>(stringListTuple2.f0, sublist));
                }
            }
        }).setParallelism(10).addSink(new SinkFunction<Tuple2<String, List<String>>>() {
            @Override
            public void invoke(Tuple2<String, List<String>> value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                System.out.println("value.f0 = " + value.f0 + "  value.f1.size()=" +value.f1.size());
            }
        }).setParallelism(1);

        env.execute("ReadTXTFilesJob");
    }




    // 假设的方法来获取目录下所有txt文件的路径
    private static List<String> getFilePaths(String dirPath, String extension) {
        List<String> paths = new ArrayList<>();
        File dir = new File(dirPath);
        if (dir.exists() && dir.isDirectory()) {
            File[] files = dir.listFiles((d, name) -> name.endsWith("." + extension));
            if (files != null) {
                for (File file : files) {
                    paths.add(file.getAbsolutePath());
                }
            }
        }
        return paths;
    }

    // 示例的FlatMapFunction用于分割每行文本
    public static final class LineSplitter implements FlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) {
            // 这里简单地输出每行内容，你可以根据需要进行处理
            out.collect(value);
        }
    }
}
